package com.atguigu.flink.exec1;

import com.atguigu.flink.pojo.UserBehavior;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashSet;

import static org.apache.flink.api.common.typeinfo.Types.*;

/**
 * Created by Smexy on 2023/1/29
 * 
 *      ①读数据
 *      ②把原始数据封装为计算的类型
 *      ③计算
 *      ④输出
 *              print
 */
public class Demo2_UV
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",3333);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        
        //env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        
        //读数据
        env.readTextFile("data/UserBehavior.csv")
           .map(new MapFunction<String, UserBehavior>()
           {
               @Override
               public UserBehavior map(String line) throws Exception {
                   String[] words = line.split(",");
                  return new UserBehavior(
                       Long.valueOf(words[0]),
                       Long.valueOf(words[1]),
                       Integer.valueOf(words[2]),
                       words[3],
                       Long.valueOf(words[4])
                   );
               }
           })
           //过滤出PV的数据
           .filter(ub -> "pv".equals(ub.getBehavior()))
           .map(ub -> ub.getUserId())
           //全局汇总
           .global()
           .process(new ProcessFunction<Long, String>()
           {
               HashSet<Long> uids = new HashSet<>();
               /*
                    上游:
                        ①source 读数据
                                    1,xxx,xxx,pv,101
                                    1,xxx,xxx,cart,102
                                    1,xxx,xxx,pv,103
                        ②Map
                                  1,xxx,xxx,pv,101  -----> UserBehavior( behavior = pv)
                                    1,xxx,xxx,cart,102 -----> UserBehavior(behavior = cart)
                                    1,xxx,xxx,pv,103 -----> UserBehavior(behavior = pv)
                        ③filter
                                  UserBehavior( behavior = pv)
                                  UserBehavior(behavior = pv)
                        ④map
                                     UserBehavior( behavior = pv)   --->1
                                     UserBehavior( behavior = pv)   --->1


                   下游:    UV: 1
                            核心处理逻辑就是去重。 可以使用Set集合实现去重！

                */
               @Override
               public void processElement(Long uid, Context ctx, Collector<String> out) throws Exception {

                   //去重
                   if (uids.add(uid)){
                       //把uv发送下游打印
                       out.collect("uv:"+uids.size());
                   }
               }
           })
           .print().setParallelism(1);


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
        
        
    }
}
